package org.future.task.redis.commit;

import org.future.task.core.executor.Executor;
import org.future.task.core.executor.FixedExecutor;
import org.future.task.core.handler.Commit;
import org.future.task.core.handler.Connection;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.LockSupport;
import java.util.function.Function;

public class RedisListCommit extends RedisCommit<Object> {

    private volatile boolean hasVal;

    private int writeBatch;

    private int localCacheSize;

    private int commitThreadSize;

    private Executor executor;

    private Connection connection;

    private ConcurrentHashMap<String, Thread> parkThread = new ConcurrentHashMap<>();

    private ConcurrentHashMap<String, LinkedBlockingQueue<Object>> lbk = new ConcurrentHashMap<>();

    public RedisListCommit(Connection connection, int localCacheSize) {
        this(connection, localCacheSize, 0);
    }

    public RedisListCommit(Connection connection, int localCacheSize, int commitThreadSize) {
        this.localCacheSize = localCacheSize;
        this.commitThreadSize = commitThreadSize;
        this.connection = connection;
    }

    public int getWriteBatch() {
        return writeBatch;
    }

    public void setWriteBatch(int writeBatch) {
        this.writeBatch = writeBatch;
    }

    @Override
    public void submit(String queueName, List<Object> dataList) throws InterruptedException {
        for (Object data : dataList) {
            submit(queueName, data);
        }
    }


    @Override
    public void submit(String queueName, Object data) throws InterruptedException {
        if (executor == null) {
            synchronized (this) {
                if (executor == null) {
                    executor = new FixedExecutor("commit-" + queueName, null, connection);
                    if (writeBatch > 0) {
                        ((FixedExecutor) executor).setWriteBatch(writeBatch);
                    }
                }
            }
        }
        LinkedBlockingQueue<Object> linkedBlockingQueue = lbk.computeIfAbsent(queueName, new Func(queueName, this));

        Thread thread = parkThread.get(queueName);
        if (thread != null) {
            LockSupport.unpark(thread);
            parkThread.remove(queueName);
        }
        linkedBlockingQueue.put(data);
    }

    class Func implements Function<String, LinkedBlockingQueue<Object>> {

        private String queueName;

        private Commit commit;

        public Func(String queueName, Commit commit) {
            this.queueName = queueName;
            this.commit = commit;
        }

        @Override
        public LinkedBlockingQueue<Object> apply(String s) {
            LinkedBlockingQueue<Object> lbk;

            if (localCacheSize > 0) {
                lbk = new LinkedBlockingQueue<>(localCacheSize);
            } else {
                lbk = new LinkedBlockingQueue<>();
            }
            //TODO 此处开启批提交
            executor.submit(queueName, commit, commitThreadSize <= 0 ? 3 : commitThreadSize);

            return lbk;
        }
    }

    @Override
    public List<Object> getBatch(String queueName, int size) throws InterruptedException {
        LinkedBlockingQueue<Object> orDefault = lbk.get(queueName);
        if (orDefault == null) {
            parkThread.put(queueName, Thread.currentThread());
            LockSupport.park();
            return null;
        }
        if (!hasVal) {
            Object take = orDefault.take();
            hasVal = true;
            return Collections.singletonList(take);
        }
        List<Object> data = new ArrayList<>();
        orDefault.drainTo(data, size);
        data.removeAll(Collections.singleton(null));
        if (data.isEmpty()) {
            hasVal = false;
            return null;
        }
        return data;
    }


}
